package com.juma.cartridge.retryque.cmd;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

import lombok.extern.slf4j.Slf4j;

import com.juma.cartridge.retryque.Conf;
import com.juma.cartridge.retryque.RetryTask;
import com.juma.cartridge.retryque.TaskFree;
import com.juma.cartridge.retryque.TaskLeaseProlong;
import com.juma.cartridge.retryque.dao.RetryQueueDao;

/**
 * 命令工作线程,专门用于回写task在数据表中的状态
 * @ClassName CommandWorker
 * @Description
 * @Author zhuangguangyou
 * @Date 2019年11月20日 下午4:49:07
 * @Version 1.0
 *
 */

@Slf4j
public class CommandWorker implements Runnable {

    private BlockingQueue<Command> commands;
    
    private RetryQueueDao dao;
    
    private Conf conf;
    
    public CommandWorker(Conf conf,BlockingQueue<Command> que,RetryQueueDao dao) {
        this.conf = conf;
        this.commands = que;
        this.dao = dao;
    }
    
    
    @Override
    public void run() {
        
        while(!Thread.currentThread().isInterrupted()) {
            Command command;
            try {
                command = commands.take();
                if(command.getAction()==Action.DELETE) {
                    doDelete(command);
                    continue;
                }
                
                if(command.getAction()==Action.UPDATE) {
                    doUpdate(command);
                    continue;
                }
                
                if(command.getAction()==Action.PROLONG) {
                    doProlong(command);
                    continue;
                }
                
                if(command.getAction()==Action.RELEASE) {
                    doRelease(command);
                    continue;
                }
            }catch(Exception e) {
                log.error("命令处理异常:"+e.getMessage(),e);
            }
        }
        
    }

    /**
     * 执行删除操作
     * @param scanner
     * @param tasks
     * @return
     */
    private boolean doDelete(Command command) {
        TaskFree free = TaskFree.of(command.getScanner(), getIds(command.getTasks()));
        return dao.delete(free)>0;
    }
    
    /**
     * 执行更新操作
     * @param command
     * @return
     */
    private boolean doUpdate(Command command) {
        return dao.updateBatch(command.getTasks())>0;
    }
    
    /**
     * 执行任务延期
     * @param command
     * @return
     */
    private boolean doProlong(Command command) {
        TaskLeaseProlong tlp = buildProlong(command);
        return dao.prolongLock(tlp)>0;
    }
    
    /**
     * 执行任务解锁
     * @param command
     * @return
     */
    private boolean doRelease(Command command) {
        TaskFree free = TaskFree.of(command.getScanner(), getIds(command.getTasks()));
        return dao.release(free)>0;
    }
    
    private List<Long> getIds(List<RetryTask> tasks) {
        List<Long> ids = new ArrayList<Long>(tasks.size());
        for (RetryTask task:tasks) {
            ids.add(task.getId());
        }
        return ids;
    }
    
    private TaskLeaseProlong buildProlong(Command command) {
        TaskLeaseProlong tr = new TaskLeaseProlong();
        tr.setScanner(command.getScanner());
        tr.setLeaseVal(command.getProlong().getLeaseVal());
        tr.setProlongVal(conf.getLockSeconds() * 1000);
        for(RetryTask task:command.getTasks()) {
            tr.addId(task.getId());    
        }
        return tr;
    }
    
}
